package com.temc.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * TODO to describe ProducerExample
 *
 * @author 辛凤文
 * @since 1.0
 */
public class ProducerExample {
    public static void main(String[] args) {
        // 配置生产者的属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "8.140.253.90:9092");  // Kafka 集群地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建 KafkaProducer 实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 创建并发送消息
            String topic = "my-topic";  // 你要发送消息的 Kafka topic
            String key = "key1";        // 消息的 key（可选）
            String value = "Hello Kafka!";  // 消息内容

            // 创建 ProducerRecord
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

            // 发送消息
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("发送失败: " + exception.getMessage());
                } else {
                    System.out.println("消息发送成功，主题: " + metadata.topic() + ", 分区: " + metadata.partition() + ", 偏移量: " + metadata.offset());
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}
